#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#include <netinet/ip_icmp.h>
#include <netdb.h>
#include <arpa/inet.h>
#include <sys/time.h>

#include "mqtt_main.h"
#include "xeLOG_internal.h"
#include "cJSON.h"
#include "his_mqtt_topic.h"
#include "event.h"
#include "../twtimer/aio-timeout.h"
#include "rbtree_queue.h"
#include "global_common.h"
struct mosquitto *mosq=NULL;
int msgCount[HS_MQTT_MSG_MODULE_ID_MAX+1]={0};
char hostip[48]="192.168.157.40";

void g_general_mqtt_publish(char *topic,const char *message);

#define ERROR (-1)
#define PACKET_SIZE     4096
#define SUCCESS 0



void connect_cbk(struct mosquitto *mosq, void *obj, int rc)
{
	int mid=0;
	STD_print("mqtt connect success");
	xeLOG_NOTICE("mqtt connect success");

	// mosquitto_subscribe(mosq, &mid, "KAREPLY", SUB_QOS);
	mosquitto_subscribe(mosq, &mid, "OPT", SUB_QOS);
	mosquitto_subscribe(mosq, &mid, "OPTREPLY", SUB_QOS);
	return;
}
void disconnect_cbk(struct mosquitto *mosq, void *obj, int result)
{
	STD_print("mqtt disconnect");
	xeLOG_NOTICE("mqtt disconnect");	
	return;
}

void publish_cbk(struct mosquitto *mosq, void *obj, int mid)
{
	return;
}

int _test_keepalive_ack(cJSON *root, char *rsp_buf)
{
	cJSON *tmp_item;

	tmp_item = cJSON_GetObjectItem(root , "event");
	if((tmp_item)&&(tmp_item->valuestring,"keepalive")){

	}
	return 0;

}


static void proc_timeout(void* param)
{
	g_general_mqtt_publish("OPTREPLY","{\"id\":\"123\",\
		\"sn\":\"222\",\"event\":\"setallosd\",\"status\":\"408\"}");
}
struct aio_timeout_t test_res;
uint64_t timeout_ms = 0;
int __proc_opttest(cJSON *root, char *rsp_buf)
{
	cJSON *tmp_item = NULL;
	char *out = NULL;

	tmp_item = cJSON_GetObjectItem(root , "item");

	return 0;
}
extern int tb_msg_release(tb_msg_t * p_msg);
extern rbtree_queue_t g_rb_queue_msg;
int __proc_opttest_ack(cJSON *root, char *rsp_buf)
{
	cJSON *tmp_item;

	tmp_item = cJSON_GetObjectItem(root , "sid");

	if(tmp_item && tmp_item->valuestring){
		RB_QUEUE_LOCK(&g_rb_queue_msg);
		char * data = rbtree_queue_search(&g_rb_queue_msg,tmp_item->valuestring);
		if(data){
			tb_msg_t *p_msg = (tb_msg_t *)data;
			deviceMgnt *p_devlist = (deviceMgnt *)(p_msg->priv_data);
			// p_msg->ref++;	//for dataref
			atomic_increment32(&p_msg->ref);
			STD_print("msg->ref 11:%d\n",p_msg->ref);
			RB_QUEUE_UNLOCK(&g_rb_queue_msg);
			STD_print("start to sleep\n");
			sleep(10);
			STD_print("end to sleep\n");
			if(0 ==aio_timeout_stop(&p_msg->timeout)){
				tb_msg_release(p_msg);//也许timer这不用加减

				RB_QUEUE_LOCK(&g_rb_queue_msg);
				rbtree_queue_erase(&g_rb_queue_msg,p_msg->msg_id);
				RB_QUEUE_UNLOCK(&g_rb_queue_msg);
				tb_msg_release(p_msg);
				STD_print("stop timer success\n");
			}else{
				STD_print("stop timer failed\n");
			}
			STD_print("msg->ref:%d\n",p_msg->ref);
			tb_msg_release(p_msg);

			STD_print("mqtt recv found,data:%p,%s,ip:%s, ref:%d",
				data,tmp_item->valuestring,p_devlist->ip,p_msg->ref);
			//for release rb_queue after proc

		}else{
			STD_print("mqtt recv not found");
			RB_QUEUE_UNLOCK(&g_rb_queue_msg);
		}
	}


	return 0;
}

int _do_process_json(char *topic, char *payload)
{
	int ret = -1;
	cJSON *root=NULL;
	char rsp_buf[256]={0},rsp_topic[32]={0};;
	if(!topic || !payload){
		xeLOG_ERR("Invalid param.");
		STD_print("Invalid param.");
		return -1;
	}
	STD_print("topic:%s payload:%s",topic,payload);
	memset(rsp_buf, 0, sizeof(rsp_buf));
	root = cJSON_Parse(payload);
	if(!root){
		xeLOG_ERR("parse JSON failed\n");
		return -1;
	}
	if(!strcmp(topic,"KAREPLY")){
		ret = _test_keepalive_ack(root,rsp_buf);
	}
	if(!strcmp(topic,"OPT")){
		ret = __proc_opttest(root,rsp_buf);
		sprintf(rsp_topic, "%s", "OPTREPLY");
	}
	if(!strcmp(topic,"OPTREPLY")){
		ret = __proc_opttest_ack(root,rsp_buf);
	}
	if(strlen(rsp_topic) && strlen(rsp_buf)){
		mosquitto_publish(mosq, NULL, rsp_topic, strlen(rsp_buf), rsp_buf, PUB_QOS, false);
	}
	if(root)
		cJSON_Delete(root);
	return ret;
}

void message_cbk(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
{
	_do_process_json(msg->topic, (char*)msg->payload);
	return;
}

void g_general_mqtt_publish(char *topic,const char *message)
{
	mosquitto_publish(mosq, NULL, topic, strlen(message), message, PUB_QOS, false);
}


void *hi3518E_mqtt_thread(void *arg)
{
	struct mosquitto *mosq=(struct mosquitto *)arg;
	prctl(PR_SET_NAME, __FUNCTION__);
	pthread_detach(pthread_self());

	mosquitto_connect_callback_set(mosq, connect_cbk);
	mosquitto_disconnect_callback_set(mosq, disconnect_cbk);
	mosquitto_publish_callback_set(mosq, publish_cbk);
	mosquitto_message_callback_set(mosq, message_cbk);

	mosquitto_username_pw_set(mosq, MQTT_USERNAME, MQTT_PASSWORD);

	mosquitto_connect(mosq, hostip, MQTT_PORT, 10);

	mosquitto_loop_forever(mosq, 10, 1);

	mosquitto_destroy(mosq);
	mosquitto_lib_cleanup();
	STD_print("hi3518E_mqtt_thread exited");
	xeLOG_NOTICE("hi3518E_mqtt_thread exited");

	pthread_exit(NULL);

	return NULL;
}
#if 0
int _hi3518E_send_heartbeat_msg(int chpid_id)
{
	cJSON *root;
	char *out = NULL;
	//unsigned int id_val;
	char item[64];
	char topic[128]={0};

	root = cJSON_CreateObject();
	if(!root){
		return -1;
	}

	hs_mqtt_msgid_update(HS_MQTT_MSG_MODULE_ID_AV,&g_powermgt_res.child_cpus[chpid_id].heartbeat_id);
	memset(item, 0 ,sizeof(item));
	snprintf(item,sizeof(item),"%d",g_powermgt_res.child_cpus[chpid_id].heartbeat_id);
	cJSON_AddStringToObject(root,"id",item);
	cJSON_AddStringToObject(root,"method","POST");
	cJSON_AddStringToObject(root,"item","heartbeat");

	memset(item, 0 ,sizeof(item));
	snprintf(item,sizeof(item),"0x%d",chpid_id);
	cJSON_AddStringToObject(root,"chipid",item);

	out=cJSON_Print(root);
	xeLOG_DEBUG("_hi3518E_send_heartbeat_msg :%s\n%s",topic,out);
	g_general_mqtt_publish(TOPIC_AV_HEARTBEAT,out);
	if(out)
		free(out);
	cJSON_Delete(root);
	return 0;
}
#endif
void hi3518E_sleep(int seconds)
{
	struct timeval tv;
	gettimeofday(&tv,NULL);
	tv.tv_sec = seconds;
	tv.tv_usec = 0;
	select(0,NULL,NULL,NULL,&tv);
}





int hs_mqtt_init(void)
{
	int ret=0;
	pthread_t pid_mqtt=-1;

	mosquitto_lib_init();
	mosq = mosquitto_new("test", true, NULL);
	if((ret=pthread_create(&pid_mqtt, NULL, (void*)hi3518E_mqtt_thread, (void*)mosq)) != 0){
		xeLOG_ERR("create mqtt thread failed\n");
		STD_print("create mqtt thread failed\n");
		goto mqtt_exit;
	}


	return ret;

mqtt_exit:
	mosquitto_destroy(mosq);
	mosquitto_lib_cleanup();

	return ret;
}
